package com.sunday.common.leaf.snowflake;

import com.sunday.common.core.gson.GsonUtils;
import com.sunday.common.leaf.snowflake.constants.SnowflakeConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.data.Stat;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.regex.Pattern;

/**
 * 分布式ID节点心跳监视器
 * 首先需要注意的是，这种情况其实有一个问题,
 * 当不同的节点在启动
 *
 * @author zsy
 * @since 2022/10/24
 */
@Slf4j
public class SnowflakeZookeeperMonitor {

    private String address;

    private int port;

    /**
     * zk数据源
     */
    private CuratorFramework curator;

    private SnowflakeZookeeperMonitor(String address, int port) {
        this.address = address;
        this.port = port;
        this.curator = createWithOptions(address, new RetryUntilElapsed(1000, 3), 10000, 6000);
        this.curator.start();
    }

    private CuratorFramework createWithOptions(String address, RetryPolicy retryPolicy, int connectionTimeoutMS, int sessionTimeoutMS) {
        return CuratorFrameworkFactory.builder().connectString(address)
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(connectionTimeoutMS)
                .sessionTimeoutMs(sessionTimeoutMS)
                .build();
    }

    /**
     * 心跳检测
     * 更新时间小于 5000 毫秒，视为存活， 固定为 3000 毫秒一次 zk 通讯交互
     *
     * @param endPoint
     * @return {@link boolean}
     */
    private boolean heartbeat(SnowflakeZookeeperHolder.Endpoint endPoint) {
        return System.currentTimeMillis() - endPoint.timestamp() < 5000;
    }

    /**
     * 获取存活的节点ID
     *
     * @return {@link Set< Integer>}
     */
    public Set<Integer> getAliveNode() {
        return getAliveNode(port);
    }

    /**
     * 根据指定端口获取存活的节点ID
     *
     * @param port
     * @return {@link Set< Integer>}
     */
    public Set<Integer> getAliveNode(int port) {
        Set<Integer> nodes = new HashSet<>();
        try {
            Stat stat = curator.checkExists().forPath(SnowflakeConstants.PATH_FOREVER);
            // 业务节点存在
            if (stat != null) {
                Pattern pattern = Pattern.compile("^(\\d+\\.\\d+\\.\\d+\\.\\d+)(:" + port + "_)");
                // 获取根路径下的所有子路径信息 /snowflake/sunday-leaf-starter/forever
                List<String> keys = curator.getChildren().forPath(SnowflakeConstants.PATH_FOREVER);
                // 根据子路径依次查找对应的值
                // 例如: /snowflake/sunday-leaf-starter/forever/10.192.xxx.xx:15000_0000000000 -> {"ip":"10.192.xxx.xx","port":"15000","timestamp":1666670143290}
                for (String key : keys) {
                    if (pattern.matcher(key).find()) {
                        byte[] bytes = curator.getData().forPath(SnowflakeConstants.PATH_FOREVER + "/" + key);
                        //筛选有效存活节点
                        if (heartbeat(GsonUtils.DEFAULT.fromJson(new String(bytes), SnowflakeZookeeperHolder.Endpoint.class))) {
                            nodes.add(Integer.parseInt(key.split("_")[1]));
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("the business sentinel monitors for survivor node exceptions:{}", e.toString(), e);
        }
        return nodes;
    }

    public static SnowflakeZookeeperMonitor of(String address, int port) {
        return new SnowflakeZookeeperMonitor(address, port);
    }

}
